package shareVal

import org.apache.spark.rdd.RDD
import org.apache.spark.{Accumulator, SparkConf, SparkContext}
import org.junit.{After, Before, Test}


/**
 * 算子练习
 */
class AccumulaterTest {
	val conf: SparkConf = new SparkConf()
	var sc: SparkContext = null
	var outpath: String = "out"
	import util.MyPredef._

	@Before
	def init() {

		conf.setAppName("WordCount")
		conf.setMaster("local")
		sc = new SparkContext(conf)

		outpath.delete
	}

	@After
	def after() {
		sc.stop()
	}


	/**
	 * Accumulator : 累加器
	 */
	@Test
	def accumulaterTest(): Unit = {

		val sum: Accumulator[Int] = sc.accumulator(0)

		val ins: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
		ins.foreach(sum.add)

		println(sum)
	}

	/**
	 * Accumulator : 自定义累加器
	 */
	@Test
	def customizeAccumulater(): Unit = {

		val accum = new LogAccumulator
		sc.register(accum, "logAccum")

		val sum = sc.parallelize(Array("1", "2a", "3", "4b", "5", "6", "7cd", "8", "9"), 2).filter(line => {
			val pattern = """^-?(\d+)"""
			val flag = line.matches(pattern)
			if (!flag) {
				accum.add(line)
			}
			flag
		}).map(_.toInt).reduce(_ + _)

		println("sum: " + sum)
		val value: java.util.Set[String] = accum.value
//		for (v <- value) print(v + "")
		println(accum.value)

	}

}


class LogAccumulator extends org.apache.spark.util.AccumulatorV2[String, java.util.Set[String]] {
	private val _logArray: java.util.Set[String] = new java.util.HashSet[String]()

	override def isZero: Boolean = {
		_logArray.isEmpty
	}

	override def reset(): Unit = {
		_logArray.clear()
	}

	override def add(v: String): Unit = {
		_logArray.add(v)
	}

	override def merge(other: org.apache.spark.util.AccumulatorV2[String, java.util.Set[String]]): Unit = {
		other match {
			case o: LogAccumulator => _logArray.addAll(o.value)
		}

	}

	override def value: java.util.Set[String] = {
		java.util.Collections.unmodifiableSet(_logArray)
	}

	override def copy():org.apache.spark.util.AccumulatorV2[String, java.util.Set[String]] = {
		val newAcc = new LogAccumulator()
		_logArray.synchronized{
			newAcc._logArray.addAll(_logArray)
		}
		newAcc
	}
}




